<?php

declare(strict_types=1);

namespace App\Amqp;

use App\Exception\BusinessException;
use Hyperf\Amqp\Builder;
use Hyperf\Di\Annotation\AnnotationCollector;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

/**
 * 1、对queue设置最大过期时间
 * 2、对发送的每个message设置过期时间（在发送时可自定义延迟）
 *
 * Class DelayProducer
 * @package App\Amqp
 */
class DelayProducer extends Builder
{
    /**
     * @param DelayProducerMessage $producerMessage //消息
     * @param int $delayTime //延迟时间
     * @param bool $confirm
     * @param int $timeout //超时
     * @return bool
     * @throws \Throwable
     */
    public function produce(
        DelayProducerMessage $producerMessage,
        int $delayTime = 0,
        bool $confirm = false,
        int $timeout = 5
    ): bool {
        return retry(
            1,
            function () use ($producerMessage, $delayTime, $confirm, $timeout) {
                return $this->produceMessage($producerMessage, $delayTime, $confirm, $timeout);
            }
        );
    }


    /**
     * @param DelayProducerMessage $producerMessage
     * @param int $delayTime
     * @param bool $confirm
     * @param int $timeout
     * @return bool
     * @throws \Throwable
     */
    private function produceMessage(
        DelayProducerMessage $producerMessage,
        int $delayTime = 0,
        bool $confirm = false,
        int $timeout = 5
    ) {
        $result = false;

        $this->injectMessageProperty($producerMessage);

        if ($delayTime > 0) {
            //设置消息延迟
            $message = new AMQPMessage(
                $producerMessage->payload(),
                array_merge(
                    $producerMessage->getProperties(),
                    ['expiration' => $delayTime * 1000]
                )
            );
        } else {
            $message = new AMQPMessage($producerMessage->payload(), $producerMessage->getProperties());
        }

        $pool = $this->getConnectionPool($producerMessage->getPoolName());
        /**
         * @var \Hyperf\Amqp\Connection $connection
         */
        $connection = $pool->get();
        if ($confirm) {
            $channel = $connection->getConfirmChannel();
        } else {
            $channel = $connection->getChannel();
        }
        $channel->set_ack_handler(
            function () use (&$result) {
                $result = true;
            }
        );

        try {
            $delayExchange = $producerMessage->getExchange();
            $delayQueue = "{$producerMessage->getExchange()}_{$delayTime}";
            $delayRoutingKey = $producerMessage->getRoutingKey() . $delayTime;

            //定义延迟交换机
            $channel->exchange_declare(
                $delayExchange,
                'topic',
                false,
                true,
                false
            );
            //定义延迟队列
            $channel->queue_declare(
                $delayQueue,
                false,
                true,
                false,
                false,
                false,
                new AMQPTable(
                    [
                        "x-dead-letter-exchange" => $producerMessage->getExchange(),
                        "x-dead-letter-routing-key" => $producerMessage->getRoutingKey(),
                        "x-message-ttl" => $producerMessage->getTtl() * 1000,//设置队列最大延迟
                    ]
                )
            );
            //绑定延迟队列到交换器上
            $channel->queue_bind(
                $delayQueue,
                $delayExchange,
                $delayRoutingKey
            );
            //消息发送
            $channel->basic_publish(
                $message,
                $delayExchange,
                $delayRoutingKey
            );
            $channel->wait_for_pending_acks_returns($timeout);
        } catch (\Throwable $e) {
            //释放前重新连接
            $connection->reconnect();
            throw new BusinessException($e->getCode(), $e->getMessage());
        } finally {
            //释放
            $connection->release();
        }

        return $confirm ? $result : true;
    }

    /**
     * @param DelayProducerMessage $producerMessage
     */
    private function injectMessageProperty(DelayProducerMessage $producerMessage)
    {
        if (class_exists(AnnotationCollector::class)) {
            /**
             * @var DelayAnnotation $annotation
             */
            $annotation = AnnotationCollector::getClassAnnotation(get_class($producerMessage), DelayAnnotation::class);
            if ($annotation) {
                $annotation->routingKey && $producerMessage->setRoutingKey($annotation->routingKey);
                $annotation->exchange && $producerMessage->setExchange($annotation->exchange);
                $annotation->ttl && $producerMessage->setTtl($annotation->ttl);
            }
        }
    }
}